package com.dongnaoedu.rpc.redis.spring;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;

import static com.dongnaoedu.rpc.redis.server.HanderResult.*;
import com.dongnaoedu.rpc.redis.server.ServerMessage;
import com.dongnaoedu.rpc.redis.client.ClientMessage;
import com.dongnaoedu.rpc.redis.server.ServerAccept;

@Component
public class ProviderListener implements MessageListener {

	private final static Log log = LogFactory.getLog(ProviderListener.class);

	private final static ConcurrentHashMap<String, CompletableFuture<ServerMessage>> concurrent_task = new ConcurrentHashMap<>();

	private final static ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

	@Autowired
	StringRedisTemplate stringRedisTemplate;

	@Override
	public void onMessage(Message message, byte[] channelByte) {
		final String serverChannel = new String(channelByte);
		//TODO regular volidate channel
		String acceptMessage = new String(message.getBody());
		log.debug("accept messge :" + acceptMessage + ". serverChannel: " + serverChannel);
		ClientMessage clientMessage = JSON.parseObject(acceptMessage, ClientMessage.class);
		// create serverAccept and volidate message
		ServerAccept serverAccept = new ServerAccept(serverChannel, clientMessage);
		//TODO This taskId must verify uniqueness
		String taskId = serverAccept.messageId;
		if (concurrent_task.contains(taskId)) {
			log.debug("The current task " + taskId + "already exists.");
			return;
		} else {
			concurrent_task.put(taskId, new CompletableFuture<>());
			cachedThreadPool.submit(new ServerAcceptTask(serverAccept));
		}

		// replay
		cachedThreadPool.submit(new ServerReplyTask(taskId, serverAccept.replyChannel));
	}

	public class ServerAcceptTask implements Runnable {

		private final ServerAccept accept;

		private final String taskId;

		public ServerAcceptTask(final ServerAccept serverAccept) {
			this.accept = serverAccept;
			this.taskId = accept.messageId;
		}

		@Override
		public void run() {
			ServerMessage handerResult = reflectInvoke(taskId, accept.className, accept.methodName, accept.args);
			concurrent_task.get(taskId).complete(handerResult);
		}

	}

	public class ServerReplyTask implements Runnable {

		private final String taskId;

		private final String replyChannel;

		public ServerReplyTask(final String taskId, final String replyChannel) {
			this.taskId = taskId;
			this.replyChannel = replyChannel;
		}

		@Override
		public void run() {
			stringRedisTemplate.convertAndSend(replyChannel, getHanderResult(taskId));
		}

		private String getHanderResult(String taskId) {
			Object reslut;
			try {
				if (concurrent_task.get(taskId) != null) {
					reslut = concurrent_task.get(taskId).get();
				} else {
					throw new IllegalStateException("This task " + taskId + " has invalid");
				}
			} catch (InterruptedException | ExecutionException e) {
				log.debug("The execution result of this task " + taskId + " has reflection abnormal information."
						+ e.getCause());
				reslut = handerResult(taskId, e);
			} finally {
				concurrent_task.remove(taskId);
			}
			return reslut.toString();
		}
	}
}
